package cn.doitedu.flink.demo;

import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.ivy.osgi.filter.MultiOperatorFilter;

import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

public class  MySink extends RichSinkFunction {

    MyOperator operator = null;
    Connection conn = null;

    public MySink(MyOperator operator) throws SQLException {
        this.operator = operator;
        conn = DriverManager.getConnection("", "", "");
    }


    @Override
    public void invoke(Object value, Context context) throws Exception {

        // 拿到sql
        String sql = operator.getSql();

        // 指定表字段类型
        List<JDBCType> tableFieldsTypes = operator.getTableFieldsTypes();

        // 拆分IN数据
        Object[] values = operator.op(value);

        // 准备sql
        PreparedStatement ps = conn.prepareStatement(sql);
        for (int i=0;i < values.length;i++) {
            ps.setObject(i+1,values[i], tableFieldsTypes.get(i));
        }
        // 执行sql
        ps.execute();
    }

    /**
     * 测试代码示例
     * @param args
     */
    public static void main(String[] args) throws Exception {
        // 指定sql
        String sql = "insert into t1 (id,name,age) values(?,?,?)";

        // 指定字段类型
        ArrayList<JDBCType> lst = new ArrayList<>();
        lst.add(JDBCType.BIGINT);
        lst.add(JDBCType.VARCHAR);
        lst.add(JDBCType.INTEGER);

        // 创建IN数据解析器
        MySink ss = new MySink(new MyOperator<String>(sql,3,lst) {

            @Override
            public Object[] op(String o) {
                String[] split = o.split(",");
                return split;
            }
        });

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.enableCheckpointing(10000);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(2);

        DataStreamSource<Object> source = env.addSource(null);

        source.addSink(ss);


        env.execute("Coupon Binlog Parser");

    }


}
